package com.minelx.datamicroscope.core.query;

import com.minelx.datamicroscope.core.display.Merger;
import com.minelx.hellopandas.core.DataFrame;
import com.minelx.hellopandas.core.PandasMethodProvider;
import io.gitee.minelx.filesystem.Directory;
import io.gitee.minelx.filesystem.File;
import io.gitee.minelx.filesystem.TempPath;

import java.util.ArrayList;
import java.util.List;

import static java.util.Collections.unmodifiableList;
import static java.util.stream.Collectors.toList;

public class Resources implements AutoCloseable {

    private final List<ResourceMapping> resourceMappings;

    private final Directory storedCluesDirectory;

    private final List<TempPath> unionClueFiles;

    private final List<TempPath> resourceFiles;

    private final Directory storedCachedIdentitiesDirectory;

    private final PandasMethodProvider pandasMethodProvider;

    private Resources(List<ResourceMapping> resourceMappings,
                      Directory storedCluesDirectory,
                      Directory storedCachedIdentitiesDirectory,
                      PandasMethodProvider pandasMethodProvider) {
        this.resourceMappings = resourceMappings;
        this.storedCluesDirectory = storedCluesDirectory;
        this.storedCachedIdentitiesDirectory = storedCachedIdentitiesDirectory;
        this.pandasMethodProvider = pandasMethodProvider;

        this.unionClueFiles = new ArrayList<>();
        this.resourceFiles = new ArrayList<>();
    }

    static Resources startWith(Directory storedCluesDirectory,
                               Directory storedCachedIdentitiesDirectory,
                               PandasMethodProvider pandasMethodProvider,
                               ResourceMapping rootResourceMapping) {
        List<ResourceMapping> resourceMappings = new ArrayList<>();
        resourceMappings.add(rootResourceMapping);

        return new Resources(resourceMappings,
                storedCluesDirectory,
                storedCachedIdentitiesDirectory,
                pandasMethodProvider);
    }

    DataFrame identifySingle(String alias) {
        return resourceMappings.stream()
                .map(ResourceMapping::resource)
                .filter(which -> which.contains(alias))
                .map(resource -> resource.identify(alias))
                .findFirst().orElseThrow(() -> new IllegalArgumentException("there is no resource file contains: " + alias));
    }

    public List<ResourceMapping> resourceMappings() {
        return unmodifiableList(resourceMappings);
    }

    DataFrame makeUnionClue(List<AliasedKey> clueKeys) {
        TempPath unionClueFile = TempPath.random(storedCluesDirectory);
        // FIXME 2022/2/26 wait for fixing.      find the relative resource files to merge instead of merge all resources, there should be a resource dependency manager to deal with.
        // left-join the whole resources
        DataFrame result = Merger.leftJoin().merge(resourceMappings, unionClueFile.file().path().uri());
        // retain clue keys
        List<String> retainedColumns = clueKeys.stream().map(AliasedKey::getAlias).collect(toList());
        result.retains(retainedColumns);
        // FIXME wait for: dropping cell which contains nan possibly causes data losing
        // preprocess to identity format
        result.dropNaN(retainedColumns, "any");
        result.dropDuplicates();

        // save as union clue
        unionClueFiles.add(unionClueFile);
        return result;
    }

    File createResource(String resourceName,
                        List<AliasedKey> clueIndexes,
                        List<AliasedKey> containedKeys) {
        TempPath resourceFile = TempPath.random(storedCluesDirectory);
        resourceFiles.add(resourceFile);
        Resource resource = Resource.create(
                storedCachedIdentitiesDirectory,
                pandasMethodProvider,
                resourceFile.file().path().uri(),
                clueIndexes,
                containedKeys);
        resourceMappings.add(new ResourceMapping(resourceName, resource));
        return resourceFile.file();
    }

    DataFrame identify(List<AliasedKey> clueKeys) {
        if (clueKeys.size() == 1) {
            // identify the only connected key from the corresponding resource file
            return identifySingle(clueKeys.get(0).getAlias());
        }
        return makeUnionClue(clueKeys);
    }

    @Override
    public void close() {
        // remove all generated resource files
        resourceFiles.forEach(TempPath::close);
        // remove all generated union clue files
        unionClueFiles.forEach(TempPath::close);
        // cleanup for resources
        resourceMappings.stream()
                .map(ResourceMapping::resource)
                .forEach(Resource::close);
    }
}
